package com.bls.executor.node.sentence.simpleStmt;

import com.bls.action.BLSAction;
import com.bls.exception.MyException;
import com.bls.executor.environment.MyMQ;
import com.bls.executor.environment.threadPool.ThreadPoolGroup;
import com.bls.executor.node.script.SimpleStmt;

import com.bls.executor.parser.V3Parser;
import com.bls.executor.word.noun.Band;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.nio.charset.StandardCharsets;
import java.util.Hashtable;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;

public class ListenStmt extends SimpleStmt {

    ThreadPoolGroup threadPoolGroup = null;

    {
        try {
            threadPoolGroup = BLSAction.getBean(ThreadPoolGroup.class);
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        }
    }

    private Band band;
    private String routingKey;
    private String typesMessage;
    private IntenseStmt intenseStmt;

    public ListenStmt(V3Parser.ListenStmtContext ctx, Map<String, Object> scriptEnvironment) throws MyException {
        String bandName = ctx.place().Identifier().getText();
        band = new Band(bandName, scriptEnvironment);
        typesMessage = ctx.Identifier().getText();

        if (typesMessage.startsWith("\"")) {
            typesMessage = typesMessage.substring(1, typesMessage.length() - 1);
        } else {
            typesMessage = typesMessage.trim();
        }

        routingKey = band.getBandId() + "." + typesMessage;

        if(ctx.intenseStmt() != null && ctx.intenseStmt().runToolStmt() != null){
            if (ctx.intenseStmt().runToolStmt().notifyStmt() == null) {
                intenseStmt = new RunToolStmt(ctx.intenseStmt().runToolStmt(), scriptEnvironment);
            } else {
                intenseStmt = new NotifyStmt(ctx.intenseStmt().runToolStmt().notifyStmt(), scriptEnvironment);
            }
        }

    }

    public Map<String, Channel> runTask() {
        Map<String, Channel> map = new Hashtable<>();
        Future<?> task = null;
        try {
            ThreadPoolExecutor executor = threadPoolGroup.getListenThreadPool();
            MyMQ mq = BLSAction.getBean(MyMQ.class);
            task = executor.submit(() -> {
                System.out.println("thread id is: " + Thread.currentThread().getId());
                try {
                    String EXCHANGE_NAME = "item_topic_exchange";//交换机名称，写死
                    //todo 要监听的队列名称，可以用时间戳+随机数 item_queue
                    String queueName = "test";
                    mq.channel.queueDeclare(queueName, false, false, false, null);
                    mq.channel.queueBind(queueName, EXCHANGE_NAME, routingKey);//定义要监听的路由类型
                    String tag = mq.channel.basicConsume(queueName, true, new Consumer() {
                        @Override
                        public void handleConsumeOk(String s) {

                        }

                        @Override
                        public void handleCancelOk(String s) {
                            System.out.println("已取消mq消费者:" + s);
                        }

                        @Override
                        public void handleCancel(String s) throws IOException {

                        }

                        @Override
                        public void handleShutdownSignal(String s, ShutdownSignalException e) {

                        }

                        @Override
                        public void handleRecoverOk(String s) {

                        }

                        @Override
                        public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
                            System.out.println("收到消息为:" + new String(bytes).getBytes(StandardCharsets.UTF_8) + ",路由为:" + routingKey);
                            intenseStmt.run("");
                        }
                    });
                    map.put(tag, mq.channel);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });

        } catch (Exception e) {
            e.printStackTrace();
        }
        try {
            task.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return map;
    }

}
